import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  
public class Step2 {  

public static class MapClass extends Mapper<LongWritable, Text, Text, Text> {  
    private Text word = new Text(); 
     
	private String getWord(String string) {
		int equalIndex = string.indexOf('=');
		if (string.charAt(0)=='{') 
			return string.substring(1, equalIndex);
		else 
			return string.substring(0, equalIndex);
	}

	private String getCount(String string) {
		int equalIndex = string.indexOf('=');
		return string.substring(equalIndex+1, string.length()-1);
	}

	
    @Override 
    public void map(LongWritable key, Text value, Context context) throws IOException,  InterruptedException {  
        try {
    	if (!value.toString().isEmpty()){
        	String lineFromText = value.toString();
            String[] lineSplited_tab = lineFromText.split("\t");
            String[] lineSplited_beforeSpace = lineSplited_tab[0].split(" ");
            String middle = lineSplited_beforeSpace[0]; 
            String denominator = lineSplited_beforeSpace[1];
            String[] lineSplited_afterSpace = lineSplited_tab[1].split(" ");
            
            for (int i=0; i<lineSplited_afterSpace.length; i++){
            	String contextWord = getWord(lineSplited_afterSpace[i]);
            	String count = getCount(lineSplited_afterSpace[i]); 
            	word.set(contextWord);
            	context.write(word, new Text(middle+"="+count+" "+denominator));
            }
    	  }
		} catch (Exception e) {
				System.out.println("Error: In Step2");
		  }
    	}
      }
  
  public static class ReduceClass extends Reducer<Text,Text,Text,Text> { 
	  				
    @Override 
    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException,  InterruptedException { 
    	HashMap<String, String> map = new HashMap<String, String>(); 	
    	List<String> valuesList = new ArrayList<String>(); 	

    	for (Text value : values) { 
    		valuesList.add(value.toString());
    	}
    	if (valuesList.size()>1){
    		for (int i=0; i<valuesList.size(); i++) {
        		for (int j=i; j<valuesList.size(); j++)
        			if (i!=j){        				
        				int firstIndexEqual = valuesList.get(i).indexOf("=");
        				int firstIndexSpace = valuesList.get(i).indexOf(" ");
        				String firstPair = valuesList.get(i).substring(0, firstIndexEqual);
        				String firstVal = valuesList.get(i).substring(firstIndexSpace, valuesList.get(i).length());
        				double firstValInt = Double.parseDouble(firstVal);
        				String firstCount = valuesList.get(i).substring(firstIndexEqual+1, firstIndexSpace);
        					
        				int secondIndexEqual = valuesList.get(j).indexOf("=");
        				int secondIndexSpace = valuesList.get(j).indexOf(" ");
        				String secondPair = valuesList.get(j).substring(0, secondIndexEqual);
        				String secondVal = valuesList.get(j).substring(secondIndexSpace, valuesList.get(j).length());
        				double secondValInt = Double.parseDouble(secondVal);
        				String secondCount = valuesList.get(j).substring(secondIndexEqual+1, secondIndexSpace);
        				
        				String count = firstCount+","+secondCount;
        				double finaldenominator = firstValInt*secondValInt;
        				String val = key+"="+count;
        				String pair1 = firstPair+" "+secondPair+" "+finaldenominator;
        				String pair2 = secondPair+" "+firstPair+" "+finaldenominator;
        				if (map.containsKey(pair1)) {
        					String oldVal = map.get(pair1);
        					map.put(pair1, oldVal+" "+val);
        				}        					
        				else {
            				if (map.containsKey(pair2)) {
            					String oldVal = map.get(pair2);
            					map.put(pair2, oldVal+" "+val);
            				}   
        				    else {
        					     map.put(pair1, val);
        				    }
        				}
        			}
        		}	
        	Set<String> keys = map.keySet();
        	for (String k : keys){
        		String v = map.get(k);
        		context.write(new Text(k), new Text(v));
        	}
    	}
      }
    }
  
 public static void main(String[] args) throws Exception {
	Path input = new Path(args[0]);
	Path output = new Path(args[1]);	
    Configuration conf = new Configuration();
    
    conf.set("mapred.max.split.size","5242880");
    
    Job job = new Job(conf, "Step2"); 
    job.setOutputKeyClass(Text.class); 
    job.setOutputValueClass(Text.class); 
    
    job.setJarByClass(Step2.class); 
    job.setMapperClass(MapClass.class); 
    job.setReducerClass(ReduceClass.class); 
    job.setInputFormatClass(TextInputFormat.class);
    job.setOutputFormatClass(TextOutputFormat.class);
    
    FileInputFormat.addInputPath(job, input); 
    FileOutputFormat.setOutputPath(job, output); 
    boolean flag = job.waitForCompletion(true); 
    if(!flag){
        System.err.println("Problem with Step2");
        return;
       }
  } 
  
}